package com.lmq

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.sql.types.StructType

object inputSource {
  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)
  val spark = SparkSession.builder()
    .master("local[*]")
    .appName(this.getClass.getName)
    .getOrCreate()

  def main(args: Array[String]): Unit = {
    // Read text from socket
    val socketDF = spark
      .readStream

      /**
       * jdbc
       * json
       * parquet
       * csv
       * libsvm
       * orc
       * nativeOrc
       * socket
       * rate
       */
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()

    val b = socketDF.isStreaming // Returns true for Dataframes that have streaming sources
    socketDF.printSchema()

    // Read all the csv files written atomically in a directory
    val userSchema = new StructType().add("name", "string")
      .add("age", "integer")
    val csvDF = spark
      .readStream
      .option("sep", ";")
      .schema(userSchema)
      .csv("D:\\javaproject\\SparkStructuredStreaming\\src\\data\\inputSource")
//    csvDF.show(false)

    val query: StreamingQuery = csvDF.writeStream
      .outputMode("append")
      .format("console")
      .start()



    query.awaitTermination()




  }

}
